/*
 * 异步工作器
 *
 * 包含异步线程，生产消费缓冲区，锁和条件变量维护线程安全。
 * 异步日志器将日志交给异步工作器，并提供日志的处理接口。
 *
 **/

#pragma once

#include <vector>
#include <thread>
#include <memory>
#include <mutex>
#include <atomic>
#include <functional>
#include <condition_variable>
#include "queue.hpp"

namespace easylog
{
    template <class buffer, class value>
    class async_worker
    {
    public:
        using ptr     = std::shared_ptr<async_worker>;
        using handler = std::function<void(buffer&)>;

        enum work_type
        {
            safe = false,
            unsafe = true,
        };

    public:
        async_worker(work_type type,
                        size_t thnum, 
                        size_t qsize, 
                        size_t incr, 
                        size_t thres, 
                        const handler& handle)
            : _pdr_queue(qsize, incr, thres)
            , _csr_queue(qsize, incr, thres)
            , _type(type)
            , _handler(handle)
            , _stop(false)
        {
            run(thnum);
        }

        void run(size_t num)
        {
            while (num--)
            {
                _th_pool.emplace_back(&async_worker::routine, this);
            }
        }

        void set_thread_num(size_t num)
        {
            _th_pool.resize(num);
        }

        void set_work_type(work_type type) { _type = type; }

        void set_buffer_size(size_t cap)
        {
            _pdr_queue.set_size(cap);
            _csr_queue.set_size(cap);
        }

        void set_handler(handler handl) { _handler = handl; }

        ~async_worker() { stop(); }

        void stop()
        {
            _stop = true;
            _csr_cv.notify_all();

            for (auto& th : _th_pool)
                th.join();
        }

        void push(value&& msg)
        {
            std::unique_lock<std::mutex> lock(_mtx);

            // When buffer is full, producer will be block in safe mode,
            // until the buffer is writable
            if (_type == safe)
            {
                _pdr_cv.wait(lock, [&]{ return _pdr_queue.wrable(); });
            }
            else if (!_pdr_queue.wrable())
            {
                _pdr_queue.reserve();
            }

            // push data and notify consumers
            _pdr_queue.push(std::forward<value>(msg));
            _csr_cv.notify_all();
        }

        // consumer working routine
        // _handler is data processing method provided by the aync logger.
        void routine()
        {
            // Even if stop is true, thread should be terminated after clearing the buffer.
            while (!_stop || _csr_queue.rdable() || _pdr_queue.rdable())
            {
                do // buffer checking and swapping.
                {
                    std::unique_lock<std::mutex> lock(_mtx);

                    if (_csr_queue.rdable()) // if consumer buffer is readable, jump to handle that
                        break;

                    // otherwise, if consumer buffer is empty,
                    // consumer wait until producer buffer is readable,
                    // then swap the two buffers and consumer continue to work.
                    _csr_cv.wait(lock, [&]{ return _stop || _pdr_queue.rdable(); });
                    _pdr_queue.swap(_csr_queue);

                    // if in safe mode, notify producer continue to work,
                    // because producer buffer is empty now.
                    if (_type == safe)
                        _pdr_cv.notify_all();
                }
                while(0);

                _handler(_csr_queue); // process one message
            }
        }

    private:
        buffer _pdr_queue;
        buffer _csr_queue;
        work_type _type;
        std::mutex _mtx;
        std::condition_variable _pdr_cv;
        std::condition_variable _csr_cv;
        std::vector<std::thread> _th_pool;
        std::atomic<bool> _stop;
        handler _handler;
    };
}
